package com.example.output;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.connector.datagen.source.DataGeneratorSource;
import org.apache.flink.connector.datagen.source.GeneratorFunction;
import org.apache.flink.connector.file.sink.FileSink;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig;
import org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.DateTimeBucketAssigner;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;

import java.time.Duration;
import java.time.ZoneId;

/**
 * Created with IntelliJ IDEA.
 * ClassName: SinkToDemo
 * Package: com.example.output
 * Description:
 * User: fzykd
 *
 * @Author: LQH
 * Date: 2023-07-19
 * Time: 12:04
 */

//输出算子
public class SinkToDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //必须开启 否则一直都是 .inprogress
        env.enableCheckpointing(2000, CheckpointingMode.AT_LEAST_ONCE);
        //数据生成器参数
        //1.实现GeneratorFunction接口 重写map方法 输入类型固定是Long
        //2.第二个参数是Long类型 自动生成的数字类型序列(从1自增) 的大小范围
        //3.限速的策略 比如每秒生成几条数据
        //4.返回的类型
        DataGeneratorSource<String> dataSource = new DataGeneratorSource<>(
                new GeneratorFunction<Long, String>() {
                    @Override
                    public String map(Long aLong) throws Exception {
                        return "Number: " + aLong;
                    }
                },
                //自动生成一个Long的最大值实现一个无界流的效果
                //设置了多个并行度 最大值按范围进行划分
                Long.MAX_VALUE,
                RateLimiterStrategy.perSecond(10),
                Types.STRING
        );
        DataStreamSource<String> data =
                env.fromSource(dataSource, WatermarkStrategy.noWatermarks(), "data");

        //输出到文件系统

        final FileSink<String> bobo = FileSink
                //输出一行的方式  存储文件 路径 + 字符编码
                .<String>forRowFormat(new Path("D:\\data\\hadoop\\Flink_Word\\input"), new SimpleStringEncoder<>())
                //输出文件名的一些配置 文件名的前缀 后缀
                .withOutputFileConfig(
                        OutputFileConfig.builder()
                                .withPartPrefix("bobo")
                                .withPartSuffix(".log")
                                .build()
                )
                //文件分桶 按照目录来分
                .withBucketAssigner(new DateTimeBucketAssigner<>("yyyy-MM-dd HH", ZoneId.systemDefault()))
                //文件滚动策略 时间是10s  大小是1m
                .withRollingPolicy(
                        DefaultRollingPolicy.builder()
                                .withInactivityInterval(Duration.ofSeconds(10))
                                .withMaxPartSize(new MemorySize(1024 * 1024))
                                .build()
                ).build();

        data.sinkTo(bobo);

        env.execute();
    }
}
